/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.training.exercises.hourlytips;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.training.exercises.common.datatypes.TaxiFare;
import org.apache.flink.training.exercises.common.source.TaxiFareGenerator;
import org.apache.flink.training.exercises.common.utils.MissingSolutionException;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * The Hourly Tips exercise from the Flink training.
 *
 * <p>The task of the exercise is to first calculate the total tips collected by each driver, hour
 * by hour, and then from that stream, find the highest tip total in each hour.
 */
public class HourlyTipsExercise {

    private final SourceFunction<TaxiFare> source;
    private final SinkFunction<Tuple3<Long, Long, Float>> sink;

    /**
     * Creates a job using the source and sink provided.
     */
    public HourlyTipsExercise(
            SourceFunction<TaxiFare> source, SinkFunction<Tuple3<Long, Long, Float>> sink) {

        this.source = source;
        this.sink = sink;
    }

    /**
     * Main method.
     *
     * @throws Exception which occurs during job execution.
     */
    public static void main(String[] args) throws Exception {

        HourlyTipsExercise job =
                new HourlyTipsExercise(new TaxiFareGenerator(), new PrintSinkFunction<>());

        job.execute();
    }

    /**
     * Create and execute the hourly tips pipeline.
     *
     * @return {JobExecutionResult}
     * @throws Exception which occurs during job execution.
     */
    public JobExecutionResult execute() throws Exception {

        // set up streaming execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // start the data generator


        //设置一个小时的 水位线
        WatermarkStrategy<TaxiFare> strategy = WatermarkStrategy
                .<TaxiFare>forBoundedOutOfOrderness(Duration.ofHours(1))
                .withTimestampAssigner((fare, timestamp) -> fare.getEventTimeMillis());

        DataStream<TaxiFare> fares = env.addSource(source)
                .assignTimestampsAndWatermarks(strategy);

//        SingleOutputStreamOperator<Tuple3<Long, Long, Float>> process = fares.keyBy(o -> o.driverId)
//                .window(TumblingEventTimeWindows.of(Time.hours(1)))
//                .process(new MyWastefulMax());


        SingleOutputStreamOperator<Tuple3<Long, Long, Float>> process = fares.keyBy(o -> o.driverId)
                .process(new keyProcess(Time.hours(1)));

        DataStream<Tuple3<Long, Long, Float>> hourlyMax = process.windowAll(TumblingEventTimeWindows.of(Time.hours(1))).maxBy(2);

        // the results should be sent to the sink that was passed in
        // (otherwise the tests won't work)
        // you can end the pipeline with something like this:

        // DataStream<Tuple3<Long, Long, Float>> hourlyMax = ...
        hourlyMax.addSink(sink);

        // execute the pipeline and return the result
        return env.execute("Hourly Tips");
    }

    private class MyWastefulMax extends ProcessWindowFunction<
            TaxiFare,
            Tuple3<Long, Long, Float>,
            Long,
            TimeWindow> {
        @Override
        public void process(Long key, ProcessWindowFunction<TaxiFare, Tuple3<Long, Long, Float>, Long, TimeWindow>.Context context, Iterable<TaxiFare> elements, Collector<Tuple3<Long, Long, Float>> out) throws Exception {
            //每个窗口的最大值
            float sumOfTips = 0F;
            for (TaxiFare fare : elements) {
                sumOfTips += fare.tip;
            }
            out.collect(Tuple3.of(key, context.window().getEnd(), sumOfTips));
        }
    }


    private class keyProcess extends KeyedProcessFunction<Long, TaxiFare, Tuple3<Long, Long, Float>> {
        //单位时间
        private final long durationMsec;

        public keyProcess(Time duration) {
            this.durationMsec = duration.toMilliseconds();
        }
       // private transient MapState<Long, Float> sumOfTips;
       private transient MapState<Long, Float> sumOfTips;
        @Override
        public void open(Configuration parameters) throws Exception {
            MapStateDescriptor<Long, Float> sumDesc =
                    new MapStateDescriptor<>("sumOfTips", Long.class, Float.class);
            //sumDesc.enableTimeToLive();
            sumOfTips = getRuntimeContext().getMapState(sumDesc);
        }

        @Override
        public void processElement(TaxiFare fare, KeyedProcessFunction<Long, TaxiFare, Tuple3<Long, Long, Float>>.Context ctx, Collector<Tuple3<Long, Long, Float>> out) throws Exception {
            long eventTime = fare.getEventTimeMillis();

            TimerService timerService = ctx.timerService();

            if (eventTime < timerService.currentWatermark()) {

            } else {
                // 将 eventTime 向上取值并将结果赋值到包含当前事件的窗口的末尾时间点。
                long endOfWindow = (eventTime - (eventTime % durationMsec) + durationMsec);
                timerService.registerEventTimeTimer(endOfWindow);

                Float sum = sumOfTips.get(endOfWindow);
                if (sum == null) {
                    sum = 0.0F;
                }
                sum += fare.tip;
                sumOfTips.put(endOfWindow, sum);
            }
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<Long, TaxiFare, Tuple3<Long, Long, Float>>.OnTimerContext ctx, Collector<Tuple3<Long, Long, Float>> out) throws Exception {
            long driverId = ctx.getCurrentKey();
            // 查找刚结束的一小时结果。
            Float sumOfTips = this.sumOfTips.get(timestamp);

            Tuple3<Long, Long, Float> result = Tuple3.of(driverId, timestamp, sumOfTips);
            out.collect(result);
            this.sumOfTips.remove(timestamp);
        }
    }
}
